package cn.doitedu.smart_marketing.sources;


import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2021/7/22
 **/
public class KafkaSourceBuilder {
    Properties props;
    public KafkaSourceBuilder() {
        Config config = ConfigFactory.load();
        props = new Properties();
        props.setProperty("bootstrap.servers",config.getString("kafka.bootstrap.servers"));
        props.setProperty("auto.offset.reset", "latest");
    }

    public <T> FlinkKafkaConsumer<T> build(String topic){
        return build(topic,null);
    }


    public <T> FlinkKafkaConsumer<T> build(String topic, Properties props){
        // 如果props!=null 合并配置

        FlinkKafkaConsumer<T> consumer = new FlinkKafkaConsumer<T>(topic, (DeserializationSchema<T>) new SimpleStringSchema(),this.props);
        return consumer;
    }
}
